package com.xdja.kafka.hdfs.sink.writer.definition;

import org.apache.hadoop.conf.Configuration;
import org.apache.parquet.example.data.GroupFactory;
import org.apache.parquet.example.data.simple.SimpleGroupFactory;
import org.apache.parquet.hadoop.example.GroupWriteSupport;
import org.apache.parquet.schema.MessageType;
import org.apache.parquet.schema.MessageTypeParser;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ParquetWriterDefinition {
    private static final Logger log = LoggerFactory.getLogger(ParquetWriterDefinition.class);
    /**
     * 配置文件
     */
    public static Configuration configuration = new Configuration();
    /**
     * parquet输出流的的support
     */
    public static GroupWriteSupport groupWriteSupport;

    /**
     * group工厂
     */
    public static GroupFactory groupFactory;
    /**
     * 初始化groupWriteSupport
     */
    static {
        try {
            //定义parquet格式
            MessageType messageType = buildMessageType();
            groupFactory = new SimpleGroupFactory(messageType);
            //定义GroupWriteSupport
            groupWriteSupport = new GroupWriteSupport();
            groupWriteSupport.setSchema(messageType, configuration);
        } catch(Exception e) {
            log.error("ParquetDefinition初始化失败，原因:{}", e.getMessage(), e);
        }
    }
    /**
     * 定义消息格式
     * @return
     */
    private static MessageType buildMessageType() {
        String messageTypeDefinition =
                "message Pair {\n" +
                "   optional binary id (UTF8);\n" +
                "   optional binary policeId (UTF8);\n" +
                "   optional binary requestId (UTF8);\n" +
                "   optional binary cardNo (UTF8);\n" +
                "   optional binary sessionId (UTF8);\n" +
                "   optional binary terminalIp (UTF8);\n" +
                "   optional binary sn (UTF8);\n" +
                "   optional binary source (UTF8);\n" +
                "   optional binary logType (UTF8);\n" +
                "   optional binary bizType (UTF8);\n" +
                "   optional binary module (UTF8);\n" +
                "   optional binary params (UTF8);\n" +
                "   optional binary formatParam (UTF8);\n" +
                "   optional binary url (UTF8);\n" +
                "   optional binary uri (UTF8);\n" +
                "   optional binary content (UTF8);\n" +
                "   optional binary response (UTF8);\n" +
                "   optional binary responseType (UTF8);\n" +
                "   optional int32 responseTime;\n" +
                "   optional binary result (UTF8);\n" +
                "   optional binary sourceIp (UTF8);\n" +
                "   optional int32 sourcePort;\n" +
                "   optional binary destIp (UTF8);\n" +
                "   optional int32 destPort;\n" +
                "   required int64 time;\n" +
                "   optional binary errorCode (UTF8);\n" +
                "   optional binary errorLog (UTF8);\n" +
                "   optional binary systemId (UTF8);\n" +
                "   optional binary networkCode (UTF8);\n" +
                "   optional binary systemName (UTF8);\n" +
                "   optional binary networkName (UTF8);\n" +
                "   optional binary keyWords (UTF8);\n" +
                "}";
        MessageType messageType = MessageTypeParser.parseMessageType(messageTypeDefinition);
        return messageType;
    }
}
